package org.carl.base.driving.core;

import cn.hutool.core.util.IdUtil;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.carl.base.driving.bean.DrivingBeanConfig;
import org.carl.base.driving.enums.SendHandleStateEnum;
import org.carl.base.driving.entity.SendMessage;
import org.carl.base.driving.service.send.SendMessageService;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionSynchronizationAdapter;
import org.springframework.transaction.support.TransactionSynchronizationManager;

import java.util.Date;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;

@Slf4j
@Component
@ConditionalOnBean(value ={DrivingBeanConfig.class})
public class SendExecute {

    @Autowired
    private SendMessageService sendMessageService;

    @Autowired
    @Qualifier("driveRabbitTemplate")
    private RabbitTemplate driveRabbitTemplate;

//    @Autowired
//    @Qualifier("driveThreadPoolExecutor")
//    private ThreadPoolExecutor driveThreadPoolExecutor;

    @Transactional(rollbackFor = Exception.class)
    public void execute(String eventName, String routingKey, String data) {
        // 保存消息
        SendMessage message = expireExecute(eventName, routingKey, data, null);

        // 注册事务提交事件
        TransactionSynchronizationManager.registerSynchronization(
                new TransactionSynchronizationAdapter() {
                    @Override
                    public void afterCommit() {
                        CompletableFuture.runAsync(() -> sendMessage(message));
//                        driveThreadPoolExecutor.execute(() -> sendMessage(message));
                    }
                });
    }


    @Transactional(rollbackFor = Exception.class)
    public SendMessage expireExecute(String eventName, String routingKey, String data, Date sendAt) {
        SendMessage message = new SendMessage();
        message.setSendId(IdUtil.simpleUUID());
        message.setEventName(eventName);
        message.setData(data);
        message.setExchange(driveRabbitTemplate.getExchange());
        message.setRoutingKey(routingKey);
        message.setGmtSend(sendAt == null ? new Date() : sendAt);
        message.setHandleState(SendHandleStateEnum.WAITING_CONFIRM.getCode());
        sendMessageService.insert(message);
        return message;
    }

    /**
     * 发送消息（这里用public只是为了事务）
     *
     * @param message
     */
    public void sendMessage(SendMessage message) {
        CorrelationData correlationData = new CorrelationData();
        correlationData.setId(message.getSendId());
        driveRabbitTemplate.convertAndSend(message.getRoutingKey(), (Object) JSONObject.toJSONString(message), correlationData);
        log.info("发送消息成功，sendId：{}，routingKey：{}，eventName：{}，data：{}", message.getSendId(), message.getRoutingKey(),
                message.getEventName(), message.getData());
    }

}
